package com.byd.demo;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.util.Collections;

/**
 *
 * 此demo仅做简单演示，实际使用需要考虑项目实际情况，消费端需要做好幂等性。
 *
 * @author jinsuo.lin@definesys.com
 * @Description TODO
 * @Version 1.0.0
 * @Date 2023/3/12 17:16
 */
@Component
@Slf4j
public class DemoConsumer {

    @Value("${rocketmq.endpoint}")
    private String endpoint;
    @Value("${rocketmq.group}")
    private String group;
    @Value("${rocketmq.topic}")
    private String topic;
    @Value("${rocketmq.tag}")
    private String tag;
    private ClientServiceProvider provider;
    private ClientConfiguration clientConfiguration;

    @PostConstruct
    public void initConsumer() throws ClientException {
        provider = ClientServiceProvider.loadService();
        clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoint)
                .build();
        consume();
    }

    /**
     * 启动消费者
     * @throws ClientException
     */
    public void consume() throws ClientException {
        // 订阅消息的过滤规则，（在ECM场景，生产者只会往指定tag推送消息）
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        PushConsumer consumer = provider.newPushConsumerBuilder()
                //设置消费者分组。
                .setConsumerGroup(group)
                //设置接入点。
                .setClientConfiguration(clientConfiguration)
                //设置预绑定的订阅关系。
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                //设置消费监听器。
                .setMessageListener(messageView -> {
                    try {
                        String messageBody = getMsg(messageView.getBody());
                        log.info("msg:" + messageBody);
                        //消费消息并返回处理结果。非常重要：需要做好幂等处理！
                        DemoDto demoDto = JSON.parseObject(messageBody, DemoDto.class);
                        if (!demoDto.getSex().equals("male")) {
                            log.info("sex:" + demoDto.getSex());
                        }
                        return ConsumeResult.SUCCESS;
                    } catch (Exception e) {
                        log.error("error:", e);
                        return ConsumeResult.FAILURE;
                    }
                })
                .build();
    }

    private String getMsg(ByteBuffer buffer){
        Charset charset = Charset.forName("utf-8");
        CharBuffer charBuffer = charset.decode(buffer);
        return charBuffer.toString();
    }
}
